import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;

import java.util.Random;

/**
 * 使用蒙洛卡特方法计算圆周率
 */
public class IterativePi {

    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env=ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);
        //迭代次数
        int iterativeNum=10000000;
        Random random=new Random(1);
        IterativeDataSet<Integer> iterativeDataSet=env.fromElements(0).iterate(iterativeNum);

        DataSet<Integer> mapResult=iterativeDataSet.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer value) throws Exception {
                double x=random.nextDouble();
                double y=random.nextDouble();
                double k = x*x+y*y;
//                System.out.println("--k--：" + k);
                value+=(k<=1)?1:0;
//                System.out.println("--value--：" + value);
                return value;
            }
        });

        //迭代结束的条件
        DataSet<Integer> result=iterativeDataSet.closeWith(mapResult);
//        System.out.println("----：" + result.map());
        result.map(new MapFunction<Integer, Double>() {
            @Override
            public Double map(Integer count) throws Exception {
                System.out.println("----：" + count);
                return count/(double)iterativeNum*4;
            }
        }).print();
    }
}
